package org.kth.dks.planetlab;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Random;

import org.kth.dks.DKSCallbackInterface;
import org.kth.dks.DKSObject;
import org.kth.dks.dks_comm.DKSRef;
import org.kth.dks.dks_dht.DKSDHTCallback;
import org.kth.dks.dks_dht.DKSDHTImpl;
import org.kth.dks.dks_marshal.DKSMessage;
import org.kth.dks.dks_node.DKSNode;
import org.kth.dks.planetlab.messages.PingMsg;
import org.kth.dks.planetlab.messages.PongMsg;
import org.kth.dks.planetlab.messages.StrechMeasurementMsg;
import org.kth.dks.util.AsyncOperation;
import org.kth.dks.util.OperationType;

public class StrechMeter extends Thread implements DKSDHTCallback {

	public static final int pingTimeout = 60;

	private String hostname;

	private boolean mysqlGateway;

	private DKSDHTImpl thisNode;

	private Connection connection;

	private PreparedStatement psStrech;

	private PreparedStatement psConnectivity;

	private static Random random = new Random();

	private DKSRef myDKSRef;

	private DKSRef currentResponsible;

	private DKSRef mySQLNode;

	DKSCallbackInterface dks = null;

	public StrechMeter(String hostname, boolean mysqlGateway, DKSRef mySQLNode) {
		super();
		this.hostname = hostname;
		this.mysqlGateway = mysqlGateway;
		this.mySQLNode = mySQLNode;
		this.setName(StrechMeter.class.getName());
	}

	public void setThisDKSDHTImpl(DKSDHTImpl DKSnode) {
		this.thisNode = DKSnode;
	}

	public void run() {
		System.out.println("************ hostname=" + hostname + ", thisNode="
				+ thisNode + ", meIsMySQL=" + mysqlGateway + ", mySQLNODE="
				+ mySQLNode);

		/*
		 * only the mysql gateway connects to the database server others send
		 * measurements through DKS
		 */
		if (mysqlGateway) {
			connectToMySQL();
		}

		myDKSRef = thisNode.myDKSImpl.getDKSRef();
		long intIP = HostUtils.getDKSRefAsint(myDKSRef);

		while (true) {
			sleepRandom();

			int dksRtt, directRtt;
			do {
				dksRtt = sendRandomLookup();
			} while (currentResponsible == null);

			directRtt = sendDirectPing(currentResponsible);

			if (directRtt == -1)
				continue;

			if (mysqlGateway) {
				insertResultIntoDB(intIP, HostUtils
						.getDKSRefAsint(currentResponsible), dksRtt, directRtt,
						myDKSRef, hostname);
			} else {
				StrechMeasurementMsg msg = new StrechMeasurementMsg(hostname,
						myDKSRef, intIP, HostUtils
								.getDKSRefAsint(currentResponsible), dksRtt,
						directRtt);

				thisNode.myDKSImpl.routeAsync(mySQLNode.getID(), new DKSObject(
						msg.flatten()));
			}
		}
	}

	public int sendRandomLookup() {
		long key = (random.nextLong() % DKSNode.N);
		System.out.println("SENDING RANDOM LOOKUP (" + key + ")");
		long sentTime = System.currentTimeMillis();
		currentResponsible = thisNode.findResponsible(key);
		long receivedTime = System.currentTimeMillis();
		return (int) (receivedTime - sentTime);
	}

	public int sendDirectPing(DKSRef target) {
		long sentTime = System.currentTimeMillis();

		AsyncOperation pingOp = AsyncOperation.start(OperationType.FINDTYPE);

		PingMsg pingMsg = new PingMsg(sentTime, pingOp.getKey());

		thisNode.myDKSImpl.send(target, pingMsg);
		System.out.println("Sent PING to " + target.getIP());

		Integer rtt = null;
		try {
			rtt = (Integer) pingOp.waitOn(pingTimeout * 1000);
		} catch (Exception e) {
			pingOp.cancel();
			System.err.println("***PING TIMEOUT***");
			return -1;
		}

		if (rtt == null)
			return -1;

		return rtt.intValue();
	}

	public void pingMsgHandler(DKSRef source, PingMsg msg) {
		thisNode.myDKSImpl.send(source, new PongMsg(msg.getTimestamp(), msg
				.getMsgId()));
		System.out.println("Replied to PING from " + source.getIP());
	}

	public void pongMsgHandler(DKSRef source, PongMsg msg) {
		long rtt = System.currentTimeMillis() - msg.getTimestamp();
		AsyncOperation.complete(msg.getMsgId(), new Integer((int) rtt));
		System.out.println("Received PONG from " + source.getIP());
	}

	public void strechMeasurementMsgHandler(DKSRef source,
			StrechMeasurementMsg msg) {
		insertResultIntoDB(msg.getFromIp(), msg.getToIp(), msg.getDksRttMs(),
				msg.getIpRttMs(), source, msg.getFromHostname());
	}

	public void dhtBroadcastCallback(DKSObject value) {
	}

	public DKSMessage dhtRouteCallback(long identifier, DKSMessage msg) {
		if (msg instanceof StrechMeasurementMsg) {
			StrechMeasurementMsg strechMsg = (StrechMeasurementMsg) msg;
			insertResultIntoDB(strechMsg.getFromIp(), strechMsg.getToIp(),
					strechMsg.getDksRttMs(), strechMsg.getIpRttMs(), strechMsg
							.getFromDKSRef(), strechMsg.getFromHostname());
		} else {
			System.err.println("*********Received unknown message********");
		}
		return null;
	}

	public void dhtRouteCallbackAsync(long identifier, DKSMessage msg) {
		if (msg instanceof StrechMeasurementMsg) {
			StrechMeasurementMsg strechMsg = (StrechMeasurementMsg) msg;
			insertResultIntoDB(strechMsg.getFromIp(), strechMsg.getToIp(),
					strechMsg.getDksRttMs(), strechMsg.getIpRttMs(), strechMsg
							.getFromDKSRef(), strechMsg.getFromHostname());
		} else {
			System.err.println("*********Received unknown message********");
		}
	}

	public void insertResultIntoDB(long fromIp, long toIp, int dksRttMs,
			int ipRttMs, DKSRef from, String hostname) {
		double strech = ((double) dksRttMs) / ((double) ipRttMs);
		try {
			psStrech.setLong(1, fromIp);
			psStrech.setLong(2, toIp);
			psStrech.setInt(3, dksRttMs);
			psStrech.setInt(4, ipRttMs);
			psStrech.setDouble(5, strech);

			psStrech.executeUpdate();

			psConnectivity.setString(1, hostname);
			psConnectivity.setString(2, from.getIP());
			psConnectivity.setInt(3, from.getPort());
			psConnectivity.setInt(4, (int) from.getID());

			psConnectivity.executeUpdate();
		} catch (SQLException e) {
			e.printStackTrace();
		}
	}

	public void connectToMySQL() {
		try {
			Class.forName("com.mysql.jdbc.Driver").newInstance();

			connection = DriverManager
					.getConnection("jdbc:mysql://193.10.67.72/dks?user=dks&password=DKS");

			psConnectivity = connection.prepareStatement("INSERT INTO "
					+ "connectivity VALUES(?, ?, ?, ?, now())");

			psStrech = connection
					.prepareStatement("INSERT INTO address_strech "
							+ "VALUES(?, ?, ?, ?, ?)");
		} catch (SQLException ex) {
			System.out.println("SQLException: " + ex.getMessage());
			System.out.println("SQLState: " + ex.getSQLState());
			System.out.println("VendorError: " + ex.getErrorCode());
			System.exit(1);
		} catch (IllegalAccessException ex) {
			System.err.println("Can't load com.mysql.jdbc.Driver class!");
			System.exit(1);
		} catch (ClassNotFoundException ex) {
			System.err.println("Can't load com.mysql.jdbc.Driver class!");
			System.exit(1);
		} catch (InstantiationException ex) {
			System.err.println("Can't load com.mysql.jdbc.Driver class!");
			System.exit(1);
		}
	}

	public void sleepRandom() {
		int rand = random.nextInt(20);
		try {
			System.out.println("SLEEPING(" + rand + ") secs");
			Thread.sleep(rand * 1000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}
